package com.atcyj.gmall.realtime.dwd.db.app;

import com.atcyj.gamll.realtime.common.base.BaseSqlApp;
import com.atcyj.gamll.realtime.common.constant.Constant;
import com.atcyj.gamll.realtime.common.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @author: cyj
 * @date: 2024/8/17
 */
public class DwdTradeCartAdd extends BaseSqlApp {
    public static void main(String[] args) {
        new DwdTradeCartAdd().start(10013, 4, "dwd_trade_card_add");
    }

    @Override
    public void handle(StreamExecutionEnvironment env, StreamTableEnvironment tableEnv, String groupId) {

        // 1. 读取kafka主题中的数据， 创建动态表
        createTopicDB(groupId, tableEnv);

        // 2. 从动态表中筛选出购物车表数据,创建动态表
        Table cartAddTable = getCartAddTable(tableEnv);

        // 3. 创建kafka输出映射，写出到kafka主题中，
        createKafkaSinkTable(tableEnv);

        // 4. 写出筛选的数据到表中，
        // Params:
        //tablePath – The path of the registered table (backed by a DynamicTableSink).
        cartAddTable.insertInto(Constant.TOPIC_DWD_TRADE_CART_ADD).execute();

    }

    private void createKafkaSinkTable(StreamTableEnvironment tableEnv) {
        tableEnv.executeSql(
                "create table " + Constant.TOPIC_DWD_TRADE_CART_ADD + "(\n" +
                        "  id string ,\n" +
                        "  user_id string ,\n" +
                        "  sku_id string ,\n" +
                        "  cart_price string ,\n" +
                        "  sku_num string ,\n" +
                        "  sku_name string ,\n" +
                        "  ts bigint \n" +
                        ")" + SQLUtil.getKafkaSinkSQL(Constant.TOPIC_DWD_TRADE_CART_ADD));
//        "WITH (\n" +
//                "  'connector' = 'kafka',\n" +
//                "  'topic' = '" + topicName + "',\n" +
//                "  'properties.bootstrap.servers' = '" + Constant.KAFKA_BROKERS + "',\n" +
//                "  'value.format' = 'json'\n" +
//                ")"
    }

    public Table getCartAddTable(StreamTableEnvironment tableEnv) {
        return tableEnv.sqlQuery(
                "select \n" +
                        "  `data`['id']  id, \n" +
                        "  `data`['user_id']  user_id, \n" +
                        "  `data`['sku_id']  sku_id, \n" +
                        "  `data`['cart_price']  cart_price, \n" +
                        "  if(`type`='insert',\n" +
                        "    `data`['sku_num'],\n" +
                        "    cast(cast(`data`['sku_num'] as bigint)-cast(`old`['sku_num'] as bigint) as string)\n" +
                        "    ) sku_num,\n" +
                        "  `data`['sku_name']  sku_name, \n" +
                        "  ts\n" +
                        " from topic_db\n" +
                        " where `database`='gmall'\n" +
                        " and `table`='cart_info'\n" +
                        " and (`type`='insert' or \n" +
                        "     (`type`='update' and `old`['sku_num'] is not null and \n" +
                        "      cast(`data`['sku_num'] as bigint)>cast(`old`['sku_num'] as bigint) ) )");
    }
}
